package com.hexinfo.dmpro.sparing.service.impl;


import cn.hutool.core.date.DateUtil;
import cn.hutool.core.io.file.FileAppender;
import cn.hutool.core.lang.UUID;
import com.hexinfo.dmpro.common.model.MessageCenter;
import com.hexinfo.dmpro.common.service.MessageCenterService;
import com.hexinfo.dmpro.common.utils.ConnExecuteUtil;
import com.hexinfo.dmpro.common.utils.ScanCommonConstants;
import com.hexinfo.dmpro.common.utils.StringUtils;
import com.hexinfo.dmpro.sparing.dto.HdfsDistcpDTO;
import com.hexinfo.dmpro.sparing.dto.TbHdfsDistcpDTO;
import com.hexinfo.dmpro.sparing.service.HdfsDistCPService;
import com.hexinfo.dmpro.sparing.service.HdfsService;
import lombok.Cleanup;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.sql.DataSource;
import java.io.File;
import java.sql.Connection;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Service
@Slf4j
public class HdfsServiceImpl implements HdfsService {

    @Value("${hdfs.heatBackup}")
    private String heatBackup;
    @Value("${heatBackup.testSyncPath}")
    private String logPath;          //日志打印路径

    @Autowired
    @Qualifier("connPoolHiveTarget")
    private DataSource connPoolHiveTarget;
    @Autowired
    @Qualifier("connPoolImpalaTarget")
    private DataSource connPoolImpalaTarget;

    @Autowired
    private HdfsDistCPService hdfsDistCPService;
    @Autowired
    private MessageCenterService messageCenterService;

    @Override
    public void hdfsDistcp(List<TbHdfsDistcpDTO> tbHdfsDistcpDTOs, String fileName, String id) {
        String logPath = path(fileName);
        FileAppender heatlog = new FileAppender(new File(logPath), 100, true);
        logLog(heatlog,"【测试同步开始，表数量："+tbHdfsDistcpDTOs.size()+"】",true);
        int i = 0;
        boolean state = true;
        for (TbHdfsDistcpDTO tbHdfsDistcpDTO : tbHdfsDistcpDTOs) {
            try{
                i++;
                //-------------------------------① 删表建表-------------------------
                Connection connHive = connPoolHiveTarget.getConnection();    //获取连接
                boolean table = createTable(connHive, tbHdfsDistcpDTO.getDbTbNameSou(), tbHdfsDistcpDTO.getDbTbNameTar(), heatlog);
                if (table){
                    //---------------------------② distcp同步------------------------
//                    String uuid = StringUtils.replace(UUID.randomUUID().toString(),"-","");
//                    hdfsDistCPService.updateStatus(uuid);     //创建判断结束的初始key
//                    int size = tbHdfsDistcpDTO.getHdfsDistcpDTOS().size();    //hdfs文件数量
//                    for (HdfsDistcpDTO hdfsDistcpDTO : tbHdfsDistcpDTO.getHdfsDistcpDTOS()) {
//                        hdfsDistCPService.execShell(hdfsDistcpDTO,uuid,heatlog);
//                    }
//                    //判断是否全部结束
//                    while (!hdfsDistCPService.getSize(uuid).equals(size)){
//                        try {
//                            Thread.sleep(3000);
//                        } catch (InterruptedException e) {
//                            e.printStackTrace();
//                        }
//                    }
//                    hdfsDistCPService.delStatus(uuid);  //移除key
//                    logLog(heatlog,"【distcp全部完成】",true);
                    //--------------------------② disctp同步-------------------
                    logLog(heatlog,tbHdfsDistcpDTO.getDbTbNameSou()+"表hdfs路径数量："+tbHdfsDistcpDTO.getHdfsDistcpDTOS().size(),true);
                    for (HdfsDistcpDTO hdfsDistcpDTO : tbHdfsDistcpDTO.getHdfsDistcpDTOS()) {
                        String shellTar = heatBackup + " " + hdfsDistcpDTO.getSouToTar();
                        //第二步：源--》目标
                        int b = execShellCode(shellTar,heatlog);
                        if (b == 999){
                            state = false;
                        }
                    }
                    //--------------------------③ 刷新hive、impala目标表--------------
                    Connection connHiveTar = connPoolHiveTarget.getConnection();
                    boolean hiveState = refreshTableHive(connHiveTar, tbHdfsDistcpDTO.getDbTbNameTar(), heatlog);
                    if (!hiveState){
                        state = false;
                    }
                    Connection connImpalaTar = connPoolImpalaTarget.getConnection();
                    boolean impalaState = refreshTableImpala(connImpalaTar, tbHdfsDistcpDTO.getDbTbNameTar(), heatlog);
                    if (!impalaState){
                        state = false;
                    }
                }else {
                    state = false;
                }
                logLog(heatlog,"【"+tbHdfsDistcpDTO.getDbTbNameSou()+"表同步完成】--"+i,true);
                logLog(heatlog,"",false);  //刷进日志
            }catch (Exception e){
                logLog(heatlog,tbHdfsDistcpDTO.getDbTbNameSou()+"表，获取连接池连接异常："+e,true);
                state = false;
            }
        }
        logLog(heatlog,"【测试同步结束】",true);
        logLog(heatlog,"",false);  //刷进日志
        //修改消息中心状态
        if (state){
            messageCenterChange(id,ScanCommonConstants.ClusterName.YWC.value,logPath);
        }else {
            messageCenterChange(id,ScanCommonConstants.ClusterName.YC.value,logPath);
        }
    }

    /**
     * 修改这一次消息中心状态为已完成
     * @return
     */
    public void messageCenterChange(String id,String state,String logPath) {
        MessageCenter mess = messageCenterService.getById(id);
        mess.setDataType(state);
        mess.setLogPath(logPath);
        messageCenterService.updateById(mess);
    }

    /**
     * 获取总日志路径
     *
     * @param fileName
     * @return
     */
    private String path(String fileName) {
        return logPath.replace("{date}", DateUtil.format(DateUtil.date(), "yyyyMMdd"))
                + fileName + ".txt";
    }

    /**
     * 源删表建表
     * @return
     */
    private boolean createTable(Connection connection, String souName, String tarName, FileAppender heatlog) {
        try{
            @Cleanup
            Statement st = connection.createStatement();
            //删表
            try{
                String dropSql = "drop table if exists {tarName}";
                String dropSql1 = dropSql.replace("{tarName}", tarName);
                logLog(heatlog,"删除表sql："+dropSql1,true);
                st.execute(dropSql1);
            }catch (Exception e){
                logLog(heatlog,"【删除表异常】："+e,true);
            }
            //建表
            String createSql = "create table {tarName} like {souName}";
            String createSql1 = createSql.replace("{tarName}", tarName)
                    .replace("{souName}", souName);
            logLog(heatlog,"建表sql："+createSql1,true);
            st.execute(createSql1);
            st.close();
            return true;
        }catch (Exception e){
            logLog(heatlog,"【建表异常】："+e,true);
            return false;
        }finally {
            try {
                if (connection != null) {
                    connection.close();
                }
            } catch (Exception e) {
                logLog(heatlog,"【关闭连接异常】："+e,true);
                return false;
            }
        }
    }

    /**
     * 刷新hive表
     * @return
     */
    private boolean refreshTableHive(Connection connection,String name,FileAppender heatlog) {
        try{
            String sql = "msck repair table {tableName}";
            String msckSql = sql.replace("{tableName}", name);
            @Cleanup
            Statement st = connection.createStatement();
            st.execute(msckSql);
            logLog(heatlog,"刷新hive表sql："+msckSql,true);
            return true;
        }catch (Exception e){
            logLog(heatlog,"【刷新hive表异常】："+e,true);
            return false;
        }finally {
            try {
                if (connection != null) {
                    connection.close();
                }
            } catch (Exception e) {
                logLog(heatlog,"【关闭hive连接异常】："+e,true);
                return false;
            }
        }
    }

    /**
     * 刷新impala表
     * @return
     */
    private boolean refreshTableImpala(Connection connection,String name,FileAppender heatlog) {
        try{
            String sql = "refresh {tableName}";
            String refreshSql = sql.replace("{tableName}", name);
            @Cleanup
            Statement st = connection.createStatement();
            st.execute(refreshSql);
            logLog(heatlog,"刷新impala表sql："+refreshSql,true);
            return true;
        }catch (Exception e){
            logLog(heatlog,"【刷新impala表异常】："+e,true);
            return false;
        }finally {
            try {
                if (connection != null) {
                    connection.close();
                }
            } catch (Exception e) {
                logLog(heatlog,"【关闭impala连接异常】："+e,true);
                return false;
            }
        }
    }

    /**
     * 执行shell脚本，只返回执行结果
     * @param pathOrCommand 脚本路径或者命令
     * @return
     */
    private int execShellCode(String pathOrCommand, FileAppender heatlog) {
        int exitValue = 0;
        try {
            Process ps = Runtime.getRuntime().exec(pathOrCommand);
            exitValue = ps.waitFor();
            logLog(heatlog,"DistCP命令：" + pathOrCommand+"  --执行结果："+exitValue,true);
        } catch (Exception e) {
            log.error("执行DistCP脚本失败命令：{}，执行DistCP脚本失败报错：{}",pathOrCommand,e);
            logLog(heatlog,"【执行DistCP错误】shell："+pathOrCommand+"---执行异常："+e,true);
            return 999;
        }
        return exitValue;
    }

    /**
     * 加锁，防止并发问题
     * @param heatlog
     * @param logStr
     * @param state
     */
    private synchronized void logLog(FileAppender heatlog, String logStr ,boolean state) {
        if (state){
            heatlog.append(logTime(logStr));
        }else {
            heatlog.flush();     //刷进日志中
        }
    }

    /**
     * 日志加日期时间
     * @param str
     * @return
     */
    private String logTime(String str) {
        return "["+DateUtil.format(DateUtil.date(), "yyyy-MM-dd HH:mm:ss")+"]" +str;
    }

}
